package com.ds.infrastructure.hyperspace.container.flow;

import com.ds.infrastructure.hyperspace.common.config.FlowHandlerConfig;
import com.ds.infrastructure.hyperspace.common.config.PipelineConfig;
import com.ds.infrastructure.hyperspace.common.flow.handler.FlowHandler;
import com.ds.infrastructure.hyperspace.common.flow.handler.PostFlowHandler;
import com.ds.infrastructure.hyperspace.common.flow.handler.PreFlowHandler;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.NettyTargetEndpoint;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.listener.DefaultHyperspaceEventListener;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.flow.operation.HyperspaceEntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.flow.operation.HyperspaceTargetFlowOperation;
import com.ds.infrastructure.hyperspace.container.flow.operation.TargetFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.HyperspaceIdGenerator;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import static com.ds.infrastructure.hyperspace.common.StateEnum.REQUEST_NOT_ACCEPTABLE;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.ASTERISK;

/**
 * @author  Weishaoying
 */
@Slf4j
public class FlowPipelineOrchestrator {

    private final NettyContainer nettyContainer;

    @Getter
    private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    @Getter
    private final TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation;

    private volatile Map<String, ConcurrentLinkedQueue<HyperspaceDuplexFlowPipeline>> pipelineMap;

    private volatile Map<String, ConcurrentLinkedQueue<HyperspaceDuplexFlowPipeline>> tempPipelineMap;

    public FlowPipelineOrchestrator(NettyContainer nettyContainer) {
        this.nettyContainer = nettyContainer;
        DefaultHyperspaceEventListener eventListener = new DefaultHyperspaceEventListener(nettyContainer);
        this.entryFlowOperation = new HyperspaceEntryFlowOperation(
                nettyContainer, eventListener, this);
        this.targetFlowOperation = new HyperspaceTargetFlowOperation(
                nettyContainer, eventListener, entryFlowOperation);
        this.pipelineMap = new ConcurrentHashMap<>();
    }

    public void initPipelines(List<PipelineConfig> pipelineConfigList, List<String> error) {
        if (CollectionUtils.isEmpty(pipelineConfigList)) {
            log.warn("Pipeline config not found!");
            return;
        }
        tempPipelineMap = new ConcurrentHashMap<>();
        for (PipelineConfig pipelineConfig : pipelineConfigList) {
            try {
                initPipeline(pipelineConfig);
            } catch (Exception e) {
                log.error("Init pileline failed!", e);
                error.add(e.getMessage());
            }
        }
        pipelineMap = tempPipelineMap;
    }

    public void registerPipelineHost(String hostName, HyperspaceDuplexFlowPipeline pipeline) {
        ConcurrentLinkedQueue<HyperspaceDuplexFlowPipeline> existsHostPipelineQueue =
                tempPipelineMap.computeIfAbsent(hostName, host -> new ConcurrentLinkedQueue<>());
        existsHostPipelineQueue.add(pipeline);
    }

    public void selectAndHandleInput(HyperspaceConnector hyperspaceConnector,
                                     HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        duplexFlowContext.addPayloadSize(entryFlowOperation.calculateMessageSize(data));
        if (duplexFlowContext.isClosed()) {
            return;
        }
        if (duplexFlowContext.getFlowHandlerPipeline() != null) {
            duplexFlowContext.getFlowHandlerPipeline().handleInput(
                    hyperspaceConnector, duplexFlowContext, data);
            return;
        }
        if (data != duplexFlowContext.getRequestStartState()) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext,
                    REQUEST_NOT_ACCEPTABLE, "Pipeline not selected!");
            return;
        }

        if (pipelineMap != null && !pipelineMap.isEmpty()) {
            ConcurrentLinkedQueue<HyperspaceDuplexFlowPipeline> pipelineQueue = null;
            if (duplexFlowContext.getHost() != null) {
                pipelineQueue = pipelineMap.get(duplexFlowContext.getHost());
            }
            if (pipelineQueue == null && duplexFlowContext.getGrpcServiceRouterName() != null) {
                pipelineQueue = pipelineMap.get(duplexFlowContext.getGrpcServiceRouterName());
            }
            if (pipelineQueue == null) {
                pipelineQueue = pipelineMap.get(ASTERISK);
            }
            if (!CollectionUtils.isEmpty(pipelineQueue)) {
                try {
                    for (HyperspaceDuplexFlowPipeline tempFlowHandlerPipeline : pipelineQueue) {
                        if (tempFlowHandlerPipeline.matchAndHandleInput(
                                hyperspaceConnector, duplexFlowContext, data)) {
                            return;
                        }
                    }
                } catch (Exception e) {
                    String errorMessage = String.format("[%s]Pipeline match and handle input failed!cause by %s", HyperspaceIdGenerator.generate(), e.toString());
                    log.error(errorMessage, e);
                    entryFlowOperation.interruptAndCloseContext(duplexFlowContext, REQUEST_NOT_ACCEPTABLE, errorMessage);
                    return;
                }

            }
        }
        entryFlowOperation.interruptAndCloseContext(duplexFlowContext, REQUEST_NOT_ACCEPTABLE, "Pipeline not found!");
    }

    private void initPipeline(PipelineConfig pipelineConfig) {
        String entryEndpointNamespace = pipelineConfig.getEntryEndpoint();
        String targetEndpointNamespace = pipelineConfig.getTargetEndpoint();
        if (!nettyContainer.isEntryEndpointExists(entryEndpointNamespace)) {
            log.error("Flow pipeline init failed, entry endpoint [{}] not found", entryEndpointNamespace);
            throw new FlowHandlerInitException("Flow pipeline init failed, entry endpoint [" +
                    entryEndpointNamespace + "] not found!");
        }
        NettyTargetEndpoint targetEndpoint = nettyContainer.getTargetEndpoint(targetEndpointNamespace);
        if (targetEndpoint == null) {
            log.error("Flow pipeline init failed, target endpoint [{}] not found", targetEndpointNamespace);
            throw new FlowHandlerInitException("Flow pipeline init failed, target endpoint [" +
                    targetEndpointNamespace + "] not found!");
        }

        HyperspaceDuplexFlowPipeline handlerPipeline =
                new HyperspaceDuplexFlowPipeline(this,
                        entryFlowOperation, targetFlowOperation, targetEndpoint);
        assemblePreFlowHandler(pipelineConfig, handlerPipeline);
        assemblePostFlowHandler(pipelineConfig, handlerPipeline);
        handlerPipeline.optimize();
    }

    private void assemblePreFlowHandler(PipelineConfig pipelineConfig,
                                        HyperspaceDuplexFlowPipeline flowHandlerPipeline) {
        List<FlowHandlerConfig> preFlowHandlerConfigList = pipelineConfig.getPreFlow();
        if (CollectionUtils.isEmpty(preFlowHandlerConfigList)) {
            return;
        }
        for (FlowHandlerConfig preFlowHandlerConfig : preFlowHandlerConfigList) {
            String preHandlerName = preFlowHandlerConfig.getHandler();
            FlowHandler extendedPreFlowHandler = nettyContainer.getFlowHandler(preHandlerName);
            if (!(extendedPreFlowHandler instanceof PreFlowHandler)) {
                throw new FlowHandlerInitException("Flow handler init failed, pre flow handler [" +
                        preHandlerName + "] invalid!");
            }
            Class<? extends FlowHandler> flowHandlerClass = extendedPreFlowHandler.getClass();
            PreFlowHandler tempFlowHandler;
            try {
                tempFlowHandler = (PreFlowHandler) flowHandlerClass.newInstance();
                String[] handlerParams = preFlowHandlerConfig.getHandlerParams();
                if (handlerParams != null && handlerParams.length > 0) {
                    tempFlowHandler.initHandlerParameters(nettyContainer, flowHandlerPipeline, handlerParams);
                }
                flowHandlerPipeline.setNextPreFlowHandler(tempFlowHandler);
            } catch (Exception e) {
                throw new FlowHandlerInitException(
                        "Flow handler init failed, occur error while instance flow handler!", e);
            }
        }
    }

    private void assemblePostFlowHandler(PipelineConfig pipelineConfig,
                                         HyperspaceDuplexFlowPipeline flowHandlerPipeline) {
        List<FlowHandlerConfig> postFlowHandlerConfigList = pipelineConfig.getPostFlow();
        if (CollectionUtils.isEmpty(postFlowHandlerConfigList)) {
            return;
        }
        for (int i = postFlowHandlerConfigList.size() - 1; i >= 0; i--) {
            FlowHandlerConfig postFlowHandlerConfig = postFlowHandlerConfigList.get(i);
            String postFlowHandlerName = postFlowHandlerConfig.getHandler();
            FlowHandler extendedPostFlowHandler = nettyContainer.getFlowHandler(postFlowHandlerName);
            if (!(extendedPostFlowHandler instanceof PostFlowHandler)) {
                throw new FlowHandlerInitException("Flow handler init failed, post flow handler [" +
                        postFlowHandlerName + "] invalid!");
            }
            Class<? extends FlowHandler> flowHandlerClass = extendedPostFlowHandler.getClass();
            try {
                PostFlowHandler tempPostFlowHandler = (PostFlowHandler) flowHandlerClass.newInstance();
                String[] handlerParams = postFlowHandlerConfig.getHandlerParams();
                if (handlerParams != null && handlerParams.length > 0) {
                    tempPostFlowHandler.initHandlerParameters(nettyContainer, flowHandlerPipeline, handlerParams);
                }
                flowHandlerPipeline.setNextPostFlowHandler(tempPostFlowHandler);
            } catch (Exception e) {
                throw new FlowHandlerInitException(
                        "Flow handler init failed, occur error while instance flow handler!");
            }
        }
    }

}
